package com.mini.grpc.transport.netty;

import com.mini.grpc.common.Metadata;
import com.mini.grpc.common.MethodDescriptor;
import com.mini.grpc.common.Status;
import com.mini.grpc.transport.ServerListener;
import com.mini.grpc.transport.ServerStream;
import com.mini.grpc.transport.ServerStreamListener;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Connection;
import io.netty.handler.codec.http2.Http2ConnectionEncoder;
import io.netty.handler.codec.http2.Http2Error;
import io.netty.handler.codec.http2.Http2Exception;
import io.netty.handler.codec.http2.Http2Flags;
import io.netty.handler.codec.http2.Http2FrameListener;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2Settings;
import io.netty.handler.codec.http2.Http2Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * NettyServerHandler处理HTTP/2帧，并将其转换为Mini-gRPC调用。
 */
public class NettyServerHandler implements Http2FrameListener {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
    
    private final Http2Connection connection;
    private final ServerListener listener;
    private final Map<Integer, NettyServerStream<?>> streams = new ConcurrentHashMap<>();
    private ChannelHandlerContext ctx;
    private Http2ConnectionEncoder encoder;
    
    /**
     * 创建一个新的NettyServerHandler
     *
     * @param connection HTTP/2连接
     * @param listener 服务端监听器
     */
    public NettyServerHandler(Http2Connection connection, ServerListener listener) {
        this.connection = connection;
        this.listener = listener;
    }
    
    /**
     * 设置上下文和编码器
     *
     * @param ctx 通道处理器上下文
     * @param encoder HTTP/2连接编码器
     */
    public void setContext(ChannelHandlerContext ctx, Http2ConnectionEncoder encoder) {
        this.ctx = ctx;
        this.encoder = encoder;
    }
    
    /**
     * 获取通道处理器上下文
     *
     * @return 通道处理器上下文
     */
    public ChannelHandlerContext getContext() {
        return ctx;
    }
    
    @Override
    public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) throws Http2Exception {
        logger.debug("=== onDataRead called: streamId={}, dataSize={}, padding={}, endOfStream={} ===", 
                    streamId, data.readableBytes(), padding, endOfStream);
        
        NettyServerStream<?> stream = streams.get(streamId);
        if (stream == null) {
            logger.error("Data received for unknown stream: {}", streamId);
            throw Http2Exception.connectionError(Http2Error.PROTOCOL_ERROR, "Data received for unknown stream: %d", streamId);
        }
        
        int processed = data.readableBytes() + padding;
        
        if (data.isReadable()) {
            // 处理数据帧
            logger.debug("Processing data for stream {}", streamId);
            stream.transportDataReceived(data, endOfStream);
        }
        
        if (endOfStream) {
            // 如果是最后一个数据帧，通知流已半关闭
            logger.debug("End of stream for stream {}", streamId);
            stream.transportHalfClosed();
        }
        
        return processed;
    }
    
    @Override
    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding, boolean endOfStream) throws Http2Exception {
        onHeadersRead(ctx, streamId, headers, 0, (short) 0, false, padding, endOfStream);
    }
    
    @Override
    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency, short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
        // 检查是否是新流
        if (!streams.containsKey(streamId)) {
            // 新流，创建一个新的服务端流
            logger.debug("Creating new stream {}", streamId);
            
            // 解析头部
            String path = headers.path() != null ? headers.path().toString() : null;
            if (path == null) {
                throw Http2Exception.connectionError(Http2Error.PROTOCOL_ERROR, "Path not found in headers");
            }
            
            // 解析方法名
            String methodName = path;
            if (methodName.startsWith("/")) {
                methodName = methodName.substring(1);
            }
            
            logger.debug("Processing method: {}", methodName);
            
            // 创建元数据
            Metadata metadata = new Metadata();
            headers.forEach(entry -> {
                String key = entry.getKey().toString();
                String value = entry.getValue().toString();
                
                // 跳过HTTP/2伪头部
                if (key.startsWith(":")) {
                    return;
                }
                
                if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
                    // 二进制头部
                    metadata.put(Metadata.Key.ofBinary(key), value.getBytes(StandardCharsets.UTF_8));
                } else {
                    // ASCII头部
                    metadata.put(Metadata.Key.of(key), value);
                }
            });
            
            // 创建方法描述符（简化版，实际应该根据服务注册表查找）
            // 这里我们使用StringMarshaller作为默认的序列化器
            @SuppressWarnings("unchecked")
            MethodDescriptor<String, String> methodDescriptor = MethodDescriptor.<String, String>newBuilder()
                .setFullMethodName(methodName)
                .setMethodType(MethodDescriptor.MethodType.UNARY) // 默认为一元调用
                .setRequestMarshaller(com.mini.grpc.serialization.StringMarshaller.INSTANCE)
                .setResponseMarshaller(com.mini.grpc.serialization.StringMarshaller.INSTANCE)
                .build();
            
            // 创建服务端流
            NettyServerStream<String> stream = new NettyServerStream<>(streamId, this, encoder);
            stream.setMarshaller(com.mini.grpc.serialization.StringMarshaller.INSTANCE);
            streams.put(streamId, stream);
            
            // 通知监听器
            try {
                @SuppressWarnings("unchecked")
                ServerStreamListener<String> streamListener = (ServerStreamListener<String>) listener.streamCreated(
                    (ServerStream<String>) stream,
                    (MethodDescriptor<String, String>) methodDescriptor,
                    metadata,
                    (InetSocketAddress) ctx.channel().remoteAddress());
                
                // 设置流监听器
                stream.setListener(streamListener);
                
                logger.debug("Stream {} created successfully", streamId);
                
                if (endOfStream) {
                    // 如果没有数据，通知流已半关闭
                    stream.transportHalfClosed();
                }
            } catch (Exception e) {
                logger.error("Failed to create stream {}", streamId, e);
                // 移除失败的流
                streams.remove(streamId);
                throw Http2Exception.connectionError(Http2Error.INTERNAL_ERROR, "Failed to create stream: " + e.getMessage());
            }
        } else {
            // 已存在的流，处理尾部头部
            NettyServerStream<?> stream = streams.get(streamId);
            if (stream != null && endOfStream) {
                logger.debug("Processing trailers for stream {}", streamId);
                
                // 处理尾部头部
                Metadata trailers = new Metadata();
                headers.forEach(entry -> {
                    String key = entry.getKey().toString();
                    String value = entry.getValue().toString();
                    
                    // 跳过HTTP/2伪头部
                    if (key.startsWith(":")) {
                        return;
                    }
                    
                    if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
                        // 二进制头部
                        trailers.put(Metadata.Key.ofBinary(key), value.getBytes(StandardCharsets.UTF_8));
                    } else {
                        // ASCII头部
                        trailers.put(Metadata.Key.of(key), value);
                    }
                });
                
                stream.transportTrailersReceived(trailers);
            }
        }
    }
    
    @Override
    public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight, boolean exclusive) throws Http2Exception {
        // 忽略优先级帧
    }
    
    @Override
    public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
        NettyServerStream<?> stream = streams.remove(streamId);
        if (stream != null) {
            // 流被重置
            Status status = Status.CANCELLED.withDescription("RST_STREAM received with error code: " + errorCode);
            stream.transportReportStatus(status);
        }
    }
    
    @Override
    public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
        // 忽略设置确认帧
    }
    
    @Override
    public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
        // 处理设置帧
    }
    
    @Override
    public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
        // 自动回复Ping
        encoder.writePing(ctx, true, data, ctx.newPromise());
        ctx.flush();
    }
    
    @Override
    public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
        // 忽略Ping确认帧
    }
    
    @Override
    public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId, Http2Headers headers, int padding) throws Http2Exception {
        // 服务端不应该收到推送承诺帧
        throw Http2Exception.connectionError(Http2Error.PROTOCOL_ERROR, "Server received PUSH_PROMISE frame");
    }
    
    @Override
    public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData) throws Http2Exception {
        // 处理GoAway帧
        logger.info("Received GOAWAY: lastStreamId={}, errorCode={}", lastStreamId, errorCode);
    }
    
    @Override
    public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement) throws Http2Exception {
        // 窗口更新由HTTP/2实现自动处理
    }
    
    @Override
    public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags, ByteBuf payload) throws Http2Exception {
        // 忽略未知帧
    }
    
    /**
     * 移除流
     *
     * @param streamId 流ID
     */
    public void removeStream(int streamId) {
        streams.remove(streamId);
    }
} 